package com.zhike.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Random;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/21
 * @Description:  根据业务过程将车辆日志数据拆分并分别写入不同topic
 *  电动模式行驶日志，混动模式行驶日志，新能源燃料模式行驶日志，电机行驶日志，充电桩充电日志，行驶充电日志，故障告警日志
 */
public class DwdApp {
    public static void main(String[] args) {
        //1.获取执行环境
        Configuration conf = new Configuration();
        //设置web端口
        conf.setInteger("rest.port", 10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        //2.消费car_log主题的数据并创建流
        KafkaSource<String> kafkaSource = KafkaUtils.getKafkaSource("car_log", "DwdApp");
        DataStreamSource<String> streamSource = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "car_source");

        //3.过滤掉非JSON格式的数据&将每行数据转换为JSON对象
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty") {
        };
        SingleOutputStreamOperator<JSONObject> jsonDS = streamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    //如果出现脏数据，输出到侧输出流
                    ctx.output(dirtyTag, value);
                }
            }
        });

        //获取侧输出流并输出
        SideOutputDataStream<String> dirtyDS = jsonDS.getSideOutput(dirtyTag);
        dirtyDS.print("出现脏数据>>>>>>>>>>>>>");

        //4.过滤不同日志并写入不同的topic
        //电动模式行驶日志侧输出流
        OutputTag<String> electricityTag = new OutputTag<>("electricity"){};
        //混动模式行驶日志侧输出流
        OutputTag<String> hybridTag = new OutputTag<>("hybrid"){};
        //新能源燃料模式行驶日志侧输出流
        OutputTag<String> fuelTag = new OutputTag<>("fuel"){};
        //充电桩充电日志侧输出流
        OutputTag<String> parking_chargingTag = new OutputTag<>("parking_charging"){};
        //行驶充电日志侧输出流
        OutputTag<String> running_chargingTag = new OutputTag<>("running_charging"){};
        //故障告警侧输出流
        OutputTag<String> alarmTag = new OutputTag<>("alarm"){};
        //其他日志侧输出流(如停车状态的日志)
        OutputTag<String> otherTag = new OutputTag<>("other"){};

        SingleOutputStreamOperator<String> processDS = jsonDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("vin");
            }
        }).process(new ProcessFunction<JSONObject, String>() {
            private ValueState<Integer> mileage_state;

            @Override
            public void open(Configuration parameters) throws Exception {
                mileage_state = getRuntimeContext().getState(new ValueStateDescriptor<>("mileage-tate", Integer.class));
            }

            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                Random random = new Random();
                //获取car_tatus
                Integer carStatus = value.getInteger("car_status");
                //获取charge_status
                Integer chargeStatus = value.getInteger("charge_status");
                //获取execution_mode
                Integer executionMode = value.getInteger("execution_mode");
                //获取alarm_level
                Integer alarmLevel = value.getInteger("alarm_level");
                //获取vin
                String vin = value.getString("vin");
                //获取timestamp
                Long timestamp = value.getLong("timestamp");

                //获取里程
                Integer mileage = value.getInteger("mileage");
                //获取状态里程
                Integer mileagePre = mileage_state.value();
                if (mileagePre == null) {
                    mileage_state.update(mileage - 30000);
                } else {
                    int temp = random.nextInt(10);
                    value.put("mileage", mileagePre + temp );
                    mileage_state.update(mileagePre + temp);
                }

                //判断日志类型
                if (carStatus == 1 && executionMode == 1) {
                    //电动模式行驶日志写入对应的侧输出流
                    ctx.output(electricityTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));
                    //获取电机列表并写出到主流
                    getMotorList(value, vin, timestamp, out);

                    //判断会否存在告警
                    check_alarmLevel(alarmLevel, ctx, value);

                    //判断是否在行驶充电
                    check_running_charging(chargeStatus, ctx, value);
                } else if (carStatus == 1 && executionMode == 2) {
                    //混动模式行驶日志写入对应的侧输出流
                    ctx.output(hybridTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));
                    //获取电机列表并写出到主流
                    getMotorList(value, vin, timestamp, out);

                    //判断会否存在告警
                    check_alarmLevel(alarmLevel, ctx, value);

                    //判断是否在行驶充电
                    check_running_charging(chargeStatus, ctx, value);
                } else if (carStatus == 1 && executionMode == 3) {
                    //新能源燃料模式行驶日志写入对应的侧输出流
                    ctx.output(fuelTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));

                    //获取电机列表并写出到主流
                    getMotorList(value, vin, timestamp, out);

                    //判断会否存在告警
                    check_alarmLevel(alarmLevel, ctx, value);

                    //判断是否在行驶充电
                    check_running_charging(chargeStatus, ctx, value);
                } else if (carStatus == 2 && chargeStatus == 1) {
                    //充电桩充电日志写入对应的侧输出流
                    ctx.output(parking_chargingTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));
                    //获取电机列表并写出到主流
                    getMotorList(value, vin, timestamp, out);

                    //判断会否存在告警
                    check_alarmLevel(alarmLevel, ctx, value);
                } else {
                    //其他日志
                    ctx.output(otherTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));
                    //判断会否存在告警
                    check_alarmLevel(alarmLevel, ctx, value);
                }


            }

            private void check_alarmLevel(Integer alarmLevel, ProcessFunction<JSONObject, String>.Context ctx, JSONObject value) {
                if (alarmLevel > 0) {
                    //故障告警日志写入对应的侧输出流
                    ctx.output(alarmTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));
                }
            }

            private void check_running_charging(Integer chargeStatus, ProcessFunction<JSONObject, String>.Context ctx, JSONObject value) {
                if (chargeStatus == 2) {
                    ctx.output(running_chargingTag, JSON.toJSONString(value, SerializerFeature.WriteMapNullValue));
                }
            }

            private void getMotorList(JSONObject value, String vin, Long timestamp, Collector<String> out) {
                JSONArray motorList = value.getJSONArray("motor_list");
                if (motorList != null && !motorList.isEmpty()) {
                    for (int i = 0; i < motorList.size(); i++) {
                        //遍历每一个电机
                        JSONObject motor = motorList.getJSONObject(i);
                        //为每一个电机添加公共属性
                        motor.put("vin", vin);
                        motor.put("timestamp", timestamp);
                        out.collect(JSON.toJSONString(motor, SerializerFeature.WriteMapNullValue));
                    }
                }
            }
        });


        //5.获取各个侧输出流的数据
        SideOutputDataStream<String> electricityDS = processDS.getSideOutput(electricityTag);
        SideOutputDataStream<String> hybridDS = processDS.getSideOutput(hybridTag);
        SideOutputDataStream<String> fuelDS = processDS.getSideOutput(fuelTag);
        SideOutputDataStream<String> parking_chargingDS = processDS.getSideOutput(parking_chargingTag);
        SideOutputDataStream<String> running_chargingDS = processDS.getSideOutput(running_chargingTag);
        SideOutputDataStream<String> alarmDS = processDS.getSideOutput(alarmTag);
        SideOutputDataStream<String> otherDS = processDS.getSideOutput(otherTag);

        electricityDS.print("电动模式行驶日志>>>>>>>>>>>>");
        hybridDS.print("混动模式行驶日志>>>>>>>>>>>>");
        fuelDS.print("新能源燃料模式行驶日志>>>>>>>>>>>");
        parking_chargingDS.print("充电桩充电日志>>>>>>>>>>>>");
        running_chargingDS.print("行驶充电日志>>>>>>>>>>>>");
        alarmDS.print("故障告警日志>>>>>>>>>>>>");
        otherDS.print("其他日志>>>>>>>>>>>>");
        processDS.print("电机行驶日志>>>>>>>>>>>>");

        //6.将数据写入到kafka
        String motorTopic="car_motor_log";
        String electricityTopic = "car_electricity_log";
        String hybridTopic = "car_hybrid_log";
        String fuelTopic = "car_fuel_log";
        String parking_chargingTopic = "car_parking_charging_log";
        String running_chargingTopic = "car_running_charging_log";
        String alarmTopic = "car_alarm_log";
        String otherTopic = "car_other_log";

        processDS.sinkTo(KafkaUtils.getKafkaSink(motorTopic));
        electricityDS.sinkTo(KafkaUtils.getKafkaSink(electricityTopic));
        hybridDS.sinkTo(KafkaUtils.getKafkaSink(hybridTopic));
        fuelDS.sinkTo(KafkaUtils.getKafkaSink(fuelTopic));
        parking_chargingDS.sinkTo(KafkaUtils.getKafkaSink(parking_chargingTopic));
        running_chargingDS.sinkTo(KafkaUtils.getKafkaSink(running_chargingTopic));
        alarmDS.sinkTo(KafkaUtils.getKafkaSink(alarmTopic));
        otherDS.sinkTo(KafkaUtils.getKafkaSink(otherTopic));


        try {
            env.execute("DwdApp");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }

}
